#include "kafka_producer.h"

#include <memory>

KafkaProducer::KafkaProducer(const std::string &brokers) : m_producer(nullptr), m_brokers(brokers)
{
    m_eventcb = new RdKafkaEventCb();
}

KafkaProducer::~KafkaProducer()
{
    if (m_producer) {
        delete m_producer;
    }
    if (m_eventcb) {
        delete m_eventcb;
    }
}

bool KafkaProducer::init(bool               with_sasl,
                         const std::string &protocol,
                         const std::string &mechanisms,
                         const std::string &username,
                         const std::string &password,
                         bool               with_ssl,
                         const std::string &ca_location)
{
    std::string                    errstr;
    std::shared_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
    if (conf->set("bootstrap.servers", m_brokers, errstr) != RdKafka::Conf::CONF_OK) {
        fprintf(stderr, "kafka producer set bootstrap.servers failed: %s\n", errstr.c_str());
        return false;
    }
    if (conf->set("event_cb", m_eventcb, errstr) != RdKafka::Conf::CONF_OK) {
        fprintf(stderr, "kafka producer set eventcb failed : %s\n", errstr.c_str());
        return false;
    }
    if (with_sasl) {
        if (conf->set("security.protocol", protocol, errstr) != RdKafka::Conf::CONF_OK) {
            fprintf(stderr, "kafka consumer set security.protocol failed: %s\n", errstr.c_str());
            return false;
        }
        if (conf->set("sasl.mechanisms", mechanisms, errstr) != RdKafka::Conf::CONF_OK) {
            fprintf(stderr, "kafka consumer set sasl.mechanisms failed: %s\n", errstr.c_str());
            return false;
        }
        if (conf->set("sasl.username", username, errstr) != RdKafka::Conf::CONF_OK) {
            fprintf(stderr, "kafka consumer set sasl.username failed: %s\n", errstr.c_str());
            return false;
        }
        if (conf->set("sasl.password", password, errstr) != RdKafka::Conf::CONF_OK) {
            fprintf(stderr, "kafka consumer set sasl.password failed: %s\n", errstr.c_str());
            return false;
        }
    }
    if (with_ssl) {
        if (conf->set("ssl.ca.location", ca_location, errstr) != RdKafka::Conf::CONF_OK) {
            fprintf(stderr, "kafka consumer set ssl.ca.location failed: %s\n", errstr.c_str());
            return false;
        }
    }
    m_producer = RdKafka::Producer::create(conf.get(), errstr);
    if (!m_producer) {
        fprintf(stderr, "create kafka producer failed : %s\n", errstr.c_str());
        return false;
    }
    return true;
}

bool KafkaProducer::produce(std::string topic, const std::string &message)
{
    return produce(topic, (void *)message.c_str(), message.length());
}

bool KafkaProducer::produce(std::string topic, void *payload, size_t len)
{
    /**
     * RdKafka::Topic::PARTITION_UA 自动选择分区
     */
    RdKafka::ErrorCode errCode = m_producer->produce(
        topic, RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY, payload, len, nullptr, 0, 0, nullptr, nullptr);
    if (errCode != RdKafka::ERR_NO_ERROR) {
        fprintf(stderr, "failed to produce to topic : %s, errno: %d\n", topic.c_str(), errCode);
        return false;
    }
    return true;
}